// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.tools.http;

import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.nio.protocol.HttpAsyncRequestExecutor;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOReactorException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;

public class HttpClientFactory {

    private static final Logger LOGGER = LoggerFactory.getLogger(HttpClientFactory.class);


    public static HttpSender createHttpClient(TSDBConfig tsdbConfig) {
        Objects.requireNonNull(tsdbConfig);
        ConnectingIOReactor ioReactor = initIOReactorConfig(tsdbConfig);
        PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(ioReactor);
        CloseableHttpAsyncClient httpAsyncClient = createPoolingHttpClient(tsdbConfig, cm);
        HttpSender httpSender = new HttpSender(tsdbConfig, httpAsyncClient);
        return httpSender;
    }

    private static ConnectingIOReactor initIOReactorConfig(TSDBConfig tsdbConfig) {
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom().setIoThreadCount(tsdbConfig.getIoThreadCount()).build();
        ConnectingIOReactor ioReactor = null;
        try {
            ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
        } catch (IOReactorException e) {
            LOGGER.error("", e);
        }
        return ioReactor;
    }

    private static RequestConfig initRequestConfig(TSDBConfig tsdbConfig) {
        RequestConfig requestConfig = null;
        int httpConnectTimeout = tsdbConfig.getHttpConnectTimeout();
        if (httpConnectTimeout >= 0) {
            RequestConfig.Builder requestConfigBuilder = RequestConfig.custom();
            requestConfigBuilder.setConnectTimeout(httpConnectTimeout);
            requestConfigBuilder.setSocketTimeout(httpConnectTimeout);
            requestConfigBuilder.setConnectionRequestTimeout(httpConnectTimeout);
            requestConfig = requestConfigBuilder.build();
        }
        return requestConfig;
    }

    private static CloseableHttpAsyncClient createPoolingHttpClient(TSDBConfig tsdbConfig, PoolingNHttpClientConnectionManager cm) {
        int httpConnectionPool = tsdbConfig.getHttpConnectionPool();
        int httpConnectionLiveTime = tsdbConfig.getHttpConnectionLiveTime();

        RequestConfig requestConfig = initRequestConfig(tsdbConfig);
        if (httpConnectionPool > 0) {
            cm.setMaxTotal(httpConnectionPool);
            cm.setDefaultMaxPerRoute(httpConnectionPool);
            cm.closeExpiredConnections();
        }

        HttpAsyncClientBuilder httpAsyncClientBuilder = HttpAsyncClients.custom();
        httpAsyncClientBuilder.setConnectionManager(cm);
        if (requestConfig != null) {
            httpAsyncClientBuilder.setDefaultRequestConfig(requestConfig);
        }
        if (httpConnectionLiveTime > 0) {
            HttpAsyncRequestExecutor httpAsyncCallbackExecutor = new HttpAsyncRequestExecutor(httpConnectionLiveTime);
            httpAsyncClientBuilder.setEventHandler(httpAsyncCallbackExecutor);
        }
        CloseableHttpAsyncClient client = httpAsyncClientBuilder.build();
        return client;
    }

}
